package com.atguigu.gmall.realtime.test;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/5/8 21:48
 */
public class TimeJoin extends BaseSQLApp {
    public static void main(String[] args) {
        new TimeJoin().init("LeftJoin", 10000, 1, "LeftJoin");
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        
        tEnv.executeSql("create table s1(" +
                            " id string, " +
                            " name string," +
                            " ts bigint, " +
                            " et as to_timestamp_ltz(ts, 3), " +
                            " watermark for et as et " +
                            ")" + SQLUtil.getKafkaDDL("atguigu", "s1"));
    
        // 版本表
        tEnv.executeSql("create table s2(" +
                            " id string, " +
                            " age int," +
                            " ts bigint, " +
                            " et timestamp(3) METADATA FROM 'value.source.timestamp' VIRTUAL, " +
                            " primary key (id) not enforced, " +
                            " watermark for et as et " +
                            ")" + SQLUtil.getUpsertKafkaDDL("s3"));
        
        tEnv.sqlQuery("select" +
                          " s1.id id1, s2.id id2, s1.name, s2.age " +
                          "from s1 " +
                          "join s2 for system_time as of s1.et " +
                          "on s1.id=s2.id").execute().print();
        

    }
}
